package com.adcc.ags.cgw.managed.proxy;

import com.adcc.ags.cgw.configuration.GWFactory;
import com.adcc.ags.cgw.configuration.UserListFactory;
import com.adcc.ags.cgw.entity.mq.MQState;
import com.adcc.ags.cgw.entity.user.*;
import com.adcc.ags.cgw.managed.downlink.DownlinkForward;
import com.adcc.utility.log.Log;
import com.adcc.utility.mq.configuration.MQConfiguration;
import com.adcc.utility.mq.configuration.MQConfigurationFactory;
import com.adcc.utility.mq.entity.Message;
import com.adcc.utility.mq.transfer.*;
import com.adcc.utility.mq.transfer.ibm.IBMMQConnectionPool;
import com.adcc.utility.mq.transfer.ibm.IBMMQTransfer;
import com.adcc.utility.time.Time;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Strings;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import io.dropwizard.setup.Environment;

import java.util.Date;
import java.util.Map;
import java.util.Set;

/**
 * 用户代理Managed
 */
public class UserProxy extends BaseProxy implements TopicMsgListener, MQStateListener {

    // 发送GWState线程名称
    public static final String CONSTANT_HBSEND_THREAD_NAME = "Thread.UserProxy.HBSend";

    // 心跳检测线程名称
    public static final String CONSTANT_HBCHECK_THREAD_NAME = "Thread.UserProxy.HBCheck";

    // 单例对象
    private static UserProxy instance;

    // 网关模式
    private GWMode gwMode = GWMode.NORMAL;

    // MQ连接状态
    private MQState mqState = MQState.DISCONNECTED;

    // 网关配置工厂
    private GWFactory gwFactory;

    // MQTransfer
    private MQTransfer mqTransfer = new IBMMQTransfer();

    // configuration
    private MQConfiguration configuration;

    // pool
    private MQConnectionPool pool;

    // 用户信息Map
    private Map<String, UserInfo> userInfoMap = Maps.newConcurrentMap();

    // 心跳发布Runnable
    private HbSendRunnable hbSendRunnable = new HbSendRunnable();

    // 心跳检测Runnable
    private HbCheckRunnable hbCheckRunnable = new HbCheckRunnable();

    /**
     * 构造函数
     */
    private UserProxy() {

    }

    /**
     * 单例方法
     * @return
     */
    public synchronized static UserProxy getInstance() {
        if (instance == null) {
            instance = new UserProxy();
        }
        return instance;
    }

    public GWMode getGWMode(){
        return gwMode;
    }

    public void setGWMode(GWMode gwMode) {
        this.gwMode = gwMode;
    }

    public Map<String, UserInfo> getUserInfoMap() {
        return userInfoMap;
    }

    /**
     * 建立Managed
     * @param environment
     */
    public void build(Environment environment, GWFactory gwFactory) throws Exception{
        if (environment == null) {
            throw new NullPointerException("environment is null");
        }
        if (gwFactory == null) {
            throw new NullPointerException("gwFactory is null");
        }
        this.gwFactory = gwFactory;

        // 设置MQ连接参数
        configuration = MQConfigurationFactory.getInstance().createIBMMQConfiguration(gwFactory.getMqFactory().getHost(),
                gwFactory.getMqFactory().getPort(),
                gwFactory.getMqFactory().getQueueManager(),
                gwFactory.getMqFactory().getChannel());
        mqTransfer.setConfiguration(configuration);

        // 设置MQ连接池参数
        pool = MQConnectionPoolFactory.getInstance().createIBMMQConnectionPool();
        ((IBMMQConnectionPool)pool).setActiveMode(gwFactory.getMqFactory().getActiveMode());
        ((IBMMQConnectionPool)pool).setTimeout(gwFactory.getMqFactory().getTimeout());
        ((IBMMQConnectionPool)pool).setMaxConnections(gwFactory.getMqFactory().getMaxConnections());
        ((IBMMQConnectionPool)pool).setMaxIdelConnections(gwFactory.getMqFactory().getMaxIdelConnections());
        mqTransfer.setConnectionPool(pool);
        pool.init(configuration);

        // 设置MQ监听器
        mqTransfer.setTopicListener(this, gwFactory.getHeartbeatFactory().getHbRecvTopic());
        mqTransfer.setMQStateListener(this);

        // 初始化userInfoMap
        initUserInfoMap();
        environment.lifecycle().manage(this);
    }

    /**
     * 初始化UserInfoMap
     * */
    private void initUserInfoMap() {
        if (UserListFactory.getInstance().getUserList() != null) {
            userInfoMap.clear();
            for (User user : UserListFactory.getInstance().getUserList()) {
                UserInfo userInfo = new UserInfo(user.getId(), user.getName());
                userInfo.setBackup(user.isBackup());
                userInfoMap.put(user.getId(), userInfo);
            }
        }
    }

    /**
     * 更新主用UserInfoMap，如果backup为false，只更新主用UserInfo，否则只更新备用UserInfo，
     * */
    public void updateUserInfoMap(boolean backup) {
        if (UserListFactory.getInstance().getUserList() != null) {
            // 新增UserInfo
            Set<String> newKey = Sets.newHashSetWithExpectedSize(UserListFactory.getInstance().getUserList().size());
            for (User user : UserListFactory.getInstance().getUserList()) {
                if (user.isBackup() == backup) {
                    newKey.add(user.getId());
                    if (!userInfoMap.containsKey(user.getId())) {
                        UserInfo userInfo = new UserInfo(user.getId(), user.getName());
                        userInfo.setBackup(user.isBackup());
                        userInfoMap.put(user.getId(), userInfo);
                    }
                }
            }

            // 记录UserInfo的key
            Set<String> allKey = Sets.newHashSetWithExpectedSize(userInfoMap.size());
            for (Map.Entry<String, UserInfo> entry : userInfoMap.entrySet()) {
                if (entry.getValue().isBackup() == backup) {
                    allKey.add(entry.getKey());
                }
            }

            // 删除多余的UserInfo
            allKey.removeAll(newKey);
            for (String key : allKey) {
                userInfoMap.remove(key);
            }
        }
    }

    /**
     * 发布网关状态
     * @param msg
     */
    public void publishGWState(byte[] msg) {
        try {
            Message message = new Message(msg);
            ((IBMMQTransfer) mqTransfer).sendTopic(gwFactory.getHeartbeatFactory().getHbSendTopic(), message, (int) gwFactory.getHeartbeatFactory().getHbSendTimeout());
            Log.info(UserProxy.class.getName(), "publish GW state:\r\n" + message);
        } catch (Exception ex) {
            Log.error(UserProxy.class.getName(), "publishGWState() error", ex);
        }
    }

    /**
     * 处理接收用户信息
     * @param msg
     */
    public void processUserState(String msg) {
        try {

            // 格式化心跳消息
            UserInfo userInfo = new UserInfo();
            userInfo.formatJson(msg);

            // 检查发送时间是否有效
            if(Strings.isNullOrEmpty(userInfo.getSendTime())){
                return;
            }

            // 检查用户ID和IP是否有效
            if(Strings.isNullOrEmpty(userInfo.getId()) || Strings.isNullOrEmpty(userInfo.getIp())){
                return;
            }

            // 检查用户ID是否有效
            if(userInfoMap.containsKey(userInfo.getId())){

                // 更新用户IP
                String strIP = userInfoMap.get(userInfo.getId()).getIp();
                if(Strings.isNullOrEmpty(strIP)){
                    userInfoMap.get(userInfo.getId()).setIp(userInfo.getIp());
                }

                // 验证发送时间是否有效
                String strSendTime = userInfo.getSendTime();
                if(!Strings.isNullOrEmpty(strSendTime)){
                    userInfoMap.get(userInfo.getId()).setState(UserState.ONLINE);
                    userInfoMap.get(userInfo.getId()).setSendTime(strSendTime);
                }
            }
        } catch (Exception ex) {
            Log.error(UserProxy.class.getName(), "processUserState() error", ex);
        }
    }

    @Override
    public void start() {
        try {
            // 启动心跳发布线程
            hbSendRunnable = new HbSendRunnable();
            new Thread(hbSendRunnable, CONSTANT_HBSEND_THREAD_NAME).start();
            // 启动心跳检测线程
            hbCheckRunnable = new HbCheckRunnable();
            new Thread(hbCheckRunnable, CONSTANT_HBCHECK_THREAD_NAME).start();
            // 启动接收BGS状态
            mqTransfer.startAsync();
        } catch (Exception ex) {
            Log.error(UserProxy.class.getName(), "start() error",ex);
        }
    }

    @Override
    public void stop() {
        try {
            // 启动心跳发布线程
            if (hbSendRunnable != null) {
                hbSendRunnable.close();
            }
            // 启动心跳检测线程
            if (hbCheckRunnable != null) {
                hbCheckRunnable.close();
            }
            // 停止接收BGS状态
            mqTransfer.stopAsync();
            // 释放MQ连接池
            pool.dispose();
        } catch (Exception ex) {
            Log.error(UserProxy.class.getName(), "stop() error",ex);
        }
    }

    @Override
    public void onTopicMsg(String tName, Message message, Map<String, String> mqInfo) {
        try{
            if (message != null && message.getLength() > 0) {
                Log.info(UserProxy.class.getName(), "receive message from mqInfo:" + mqInfo + " topic:" + tName + " message:\r\n" + message);
                processUserState(new String(message.getContent(), 0, message.getContent().length));
            }
        }catch (Exception ex){
            Log.error(UserProxy.class.getName(), "onMessage() error",ex);
        }
    }

    @Override
    public void onState(com.adcc.utility.mq.entity.MQState state, Map<String, String> mqInfo) {
        try {
            Log.info(UserProxy.class.getName(), "mqInfo:" + mqInfo + " state:" + state);
            if (state == com.adcc.utility.mq.entity.MQState.CONNECTED) {
                mqState = MQState.CONNECTED;
            } else if(state == com.adcc.utility.mq.entity.MQState.CONNECTING) {
                mqState = MQState.CONNECTING;
            } else {
                mqState = MQState.DISCONNECTED;
            }
        }catch (Exception ex){
            Log.error(UserProxy.class.getName(), "onState() error", ex);
        }
    }

    /**
     * 心跳发送线程
     */
    class HbSendRunnable implements Runnable {

        // 线程启动标识
        private boolean isStarted;

        /**
         * 关闭线程
         */
        public void close() {
            isStarted = false;
        }

        @Override
        public void run() {
            try {
                Log.info(HbSendRunnable.class.getName(), "Thread:" + CONSTANT_HBSEND_THREAD_NAME + " start...");
                isStarted = true;
                while (isStarted) {
                    try {
                        if (UserListFactory.getInstance().getUserList().size() > 0) {
                            Map<String, Object> map = Maps.newHashMapWithExpectedSize(4);
                            if (DownlinkForward.getInstance().getAMQS().getState() == MQState.CONNECTED && mqState == MQState.CONNECTED) {
                                map.put("state", GWState.AVAILABLE);
                            } else {
                                map.put("state", GWState.UNAVAILABLE);
                            }
                            map.put("userlist", UserListFactory.getInstance().getUserList());
                            map.put("sendTime", Time.formatDateTimeToUTCString(new Date(), Time.yyyy_MM_dd_HH_mm_ss));
                            map.put("mode", gwMode);
                            publishGWState(new ObjectMapper().writeValueAsString(map).getBytes());
                        }

                        Thread.sleep(gwFactory.getHeartbeatFactory().getHbSendInterval());
                    } catch (Exception ex) {
                        Log.error(HbSendRunnable.class.getName(), "publish GW state error", ex);
                    }
                }
            } catch (Exception ex) {
                Log.error(HbSendRunnable.class.getName(), "run() error", ex);
            }
        }
    }

    /**
     * BGS用户信息检测线程
     * */
    class HbCheckRunnable implements Runnable {

        // 线程启动标识
        private boolean isStarted;

        /**
         * 关闭线程
         */
        public void close() {
            isStarted = false;
        }

        @Override
        public void run() {
            try {
                Log.info(HbCheckRunnable.class.getName(), "Thread:" + CONSTANT_HBCHECK_THREAD_NAME + " start...");
                isStarted = true;
                while (isStarted) {
                    try {
                        for (Map.Entry<String, UserInfo> entry :  userInfoMap.entrySet()) {
                            UserInfo userInfo = entry.getValue();
                            if (userInfo.getState() == UserState.ONLINE) {
                                Date sendTime = Time.formatDateTimeToDate(userInfo.getSendTime(),Time.yyyy_MM_dd_HH_mm_ss);
                                if (Math.abs(Time.getCurUTCDate().getTime() - sendTime.getTime()) > gwFactory.getHeartbeatFactory().getHbRecvExpiredTime() * 1000) {
                                    userInfo.setState(UserState.OFFLINE);
                                    userInfo.setIp("");
                                    userInfo.setSendTime("");
                                }
                            }
                        }
                        Thread.sleep(2000);
                    } catch (Exception ex) {
                        Log.error(HbCheckRunnable.class.getName(), "check BGS HeartBeat error", ex);
                    }
                }
            } catch (Exception ex) {
                Log.error(HbCheckRunnable.class.getName(), "run() error", ex);
            }
        }
    }
}
